package net.bwie.realtime.vehicle.dwd.job.one;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.utils.DateTimeUtil;
import net.bwie.realtime.utils.DorisUtil;
import net.bwie.realtime.utils.KafkaUtil;
import net.bwie.realtime.vehicle.dwd.function.IllegalVehicleFilter;
import net.bwie.realtime.vehicle.dwd.utils.PGIllegalVehicleLoader;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/*
 * TODO 违法王轨迹跟踪
 *      违法超过30次且未处理的，判定为违法王，实时追踪打印这些车辆经过的卡口
 */

/**
 * @author ASUS
 */
public class BreakTrackJob {
    public static void main(String[] args) throws Exception {

        //1环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000L);

        //2读取主题
        DataStream<String> vehicleKafkaStream = KafkaUtil.consumerKafka(env, "dwd_car_monitor");

        // 3. 转换处理（核心逻辑：过滤违法次数超30次的车辆）
        DataStream<String> filteredStream = processBreak(vehicleKafkaStream);
        filteredStream.print();

        //4sink
        DorisUtil.saveToDoris(filteredStream,
                "vehicle_realtime_report",
                "vehicle_break_track_report");

        //5. execute-执行
        env.execute("BreakTrackJob");
    }

    /**
     * 处理车辆数据流，过滤出违法次数超30次的车辆数据
     */
    private static DataStream<String> processBreak(DataStream<String> vehicleStream) {
        // 创建PG违法车辆数据加载器
        PGIllegalVehicleLoader illegalVehicleLoader = new PGIllegalVehicleLoader(
                "jdbc:postgresql://node103:5432/smart_traffic",
                "postgres",
                "123456"
        );
        // 过滤出违法车辆数据
        SingleOutputStreamOperator<String> filter = vehicleStream.filter(new IllegalVehicleFilter(illegalVehicleLoader));
        return filter.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                Long actionTime = jsonObject.getLong("actionTime");

                jsonObject.put("actionTime", DateTimeUtil.convertLongToString(actionTime, DateTimeUtil.DATE_TIME_FORMAT));
                return jsonObject.toJSONString();
            }
        });
    }

}
